package org.jdcenter.rmi.impl;

import java.util.*;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

import static org.jdcenter.rmi.impl.Glossary.*;

import org.jdcenter.rmi.abs.*;

public class HandlerChannelCache {
    private static Logger aLogger = Logger.getLogger(HandlerChannelCache.class.getCanonicalName());
    private Logger getLogger() {
        return aLogger;
    }

    private int limit;
    private final int step;
    private final long reserve;

    private Set<Integer> rejecteds;
    private AtomicInteger sequencer;
    private List<HandlerChannel> channels;
    private DelayQueue<ReservedCookie> reserveds;

    private static HandlerChannel NULL = new HandlerChannel(null) {
        public Result invoke(Argument arg) throws ChannelException {
            throw new ChannelException("NULL handler channel is never should be called");
        }
    };

    private static class ReservedCookie extends DelayedObject {
        private int cookie;

        public ReservedCookie(int cookie, long delay) {
            super(delay);
            this.cookie = cookie;
        }

        public int getCookie() {
            return cookie;
        }
    }

    public HandlerChannelCache(Scheme context) {
        this.limit = context.getProperty(HANDLER_CACHE_INITIAL_SIZE);
        this.step = context.getProperty(HANDLER_CACHE_INCREASE_STEP);
        this.reserve = context.getProperty(HANDLER_COOKIE_RESERVED_TIME);

        this.rejecteds = Collections.synchronizedSet(new HashSet<Integer>());

        this.sequencer = new AtomicInteger(0);
        this.channels = Collections.synchronizedList(new ArrayList<HandlerChannel>());

        this.reserveds = new DelayQueue<ReservedCookie>();
    }

    public HandlerChannel get(int cookie)
                throws HandlerInvalidException, HandlerRejectedException {
        assert cookie >= 0;

        if(rejecteds.contains(Integer.valueOf(cookie))) {
            String msg = handler(cookie) + " is rejected";
            getLogger().warning(msg);
            throw new HandlerRejectedException(msg);
        }

        if (cookie >= sequencer.get()) {
            String msg = handler(cookie) + " is out of range";
            getLogger().warning(msg);
            throw new HandlerInvalidException(msg);
        }

        HandlerChannel channel = channels.get(cookie);
        if (channel == NULL) {
            String msg = handler(cookie) + " is removed";
            getLogger().warning(msg);
            throw new HandlerInvalidException(msg);
        }

        return channel;
    }

    public int put(HandlerChannel channel) {
        ReservedCookie x = reserveds.poll();
        if (x != null) {
            int index = x.getCookie();
            channels.set(index, channel);
            channel.setIdentity(this, index);
            getLogger().finer("Registered " + handler(index));
            return index;
        }

        int index = 0;

        synchronized(this) {
            index = sequencer.getAndIncrement();
            channels.add(channel);
            channel.setIdentity(this, index);
        }

        if (index >= limit) {
            getLogger().warning("Handler channel cache size reached limit " + limit);
            limit += step;
        }

        getLogger().finer("Registered " + handler(index));
        return index;
    }

    public void remove(int cookie) {
        if(channels.set(cookie, NULL) == NULL) {
            getLogger().warning("Removed un-removed " + handler(cookie));
            return;
        }

        reserveds.offer(new ReservedCookie(cookie, reserve));
        getLogger().finer("Removed " + handler(cookie));
    }

    public int resume(RpcHandler handler, int cookie) throws HandlerResumeException {
        assert cookie >= 0;

        if(rejecteds.remove(Integer.valueOf(cookie))) {
            channels.get(cookie).setHandler(handler);
            getLogger().finer("Resumed " + handler(cookie));
            return cookie;
        }

        String msg = "Resumed un-rejected " + handler(cookie);
        getLogger().warning(msg);
        throw new HandlerResumeException(msg);
    }

    public void reject(int cookie) {
        rejecteds.add(Integer.valueOf(cookie));
        getLogger().finer("Rejected " + handler(cookie));
        // Trust our self
    }
}

